package com.xiao.dynamic_rule.engine;


import com.alibaba.fastjson.JSON;
import com.xiao.dynamic_rule.pojo.LogBean;
import com.xiao.dynamic_rule.pojo.ResultBean;
import com.xiao.dynamic_rule.utils.HbaseUtils;
import com.xiao.dynamic_rule.utils.KafkaUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.Map;

/**
 * 实现静态实时运营
 * <p>
 * 需求：
 * 触发条件：E事件
 * 画像属性条件：  k3=v3 , k100=v80 , k230=v360
 * 行为属性条件：  U(p1=v3,p2=v2) >= 3次 且  G(p6=v8,p4=v5,p1=v2)>=1
 * 行为次序条件：  依次做过：  W(p1=v4) ->   R(p2=v3) -> F
 */
public class RuleEngineDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //连接kafka
        FlinkKafkaConsumer<String> kafkaData = KafkaUtils.getFlinkKafkaCounsumer("hadoop101:9092,hadoop102:9092,hadoop103:9092", "latest", "applog");
        DataStreamSource<String> stringDataStreamSource = env.addSource(kafkaData);

        //将kafka中的json日志数据转换为bean
        SingleOutputStreamOperator<ResultBean> result = stringDataStreamSource
                .map(new MapFunction<String, LogBean>() {
                    @Override
                    public LogBean map(String s) throws Exception {
                        return JSON.parseObject(s, LogBean.class);
                    }
                })
                .filter(new FilterFunction<LogBean>() {
                    @Override
                    public boolean filter(LogBean logBean) throws Exception {
                        return "E".equals(logBean.getEventId());
                    }
                })
                .keyBy(key -> key.getDeviceId())
                .process(new MAProcessFunction());

        result.print();

        env.execute();
    }





    //继承kedProcess实现逻辑
    public static class MAProcessFunction extends KeyedProcessFunction<String, LogBean, ResultBean> {

        private Table user_profile;
        private ListState<LogBean> event_state;

        @Override
        public void open(Configuration parameters) throws Exception {
            // 连接hbase
            Connection hbaseConn = HbaseUtils.HbaseConnect("hadoop101:2181,hadoop102:2181,hadoop103:2181");
            user_profile = hbaseConn.getTable(TableName.valueOf("user_profile"));

            //创建 list state来存数据
            event_state = getRuntimeContext().getListState(new ListStateDescriptor<>("event_state", LogBean.class));

        }

        @Override
        public void processElement(LogBean value, KeyedProcessFunction<String, LogBean, ResultBean>.Context ctx, Collector<ResultBean> out) throws Exception {
            //将拿到的数据存入listState
            event_state.add(value);

            //画像属性条件：  k3=v3 , k100=v80 , k230=v360
            //读取画像数据,在open方法中连接,将所需的列拿出来
            Get get = new Get(Bytes.toBytes(value.getDeviceId()));
            get.addColumn(Bytes.toBytes("f"), Bytes.toBytes("k3"));
            get.addColumn(Bytes.toBytes("f"), Bytes.toBytes("k100"));
            get.addColumn(Bytes.toBytes("f"), Bytes.toBytes("k230"));

            Result user_profileData = user_profile.get(get);
            String k3 = new String(user_profileData.getValue(Bytes.toBytes("f"), Bytes.toBytes("k3")));
            String k100 = new String(user_profileData.getValue(Bytes.toBytes("f"), Bytes.toBytes("k100")));
            String k230 = new String(user_profileData.getValue(Bytes.toBytes("f"), Bytes.toBytes("k230")));

            //TODO 判断画像数据是否满足
            if ("v3".equals(k3) && "v80".equals(k100) && "v360".equals(k230)) {
                //TODO 判断 行为属性条件：  U(p1=v3,p2=v2) >= 3次 且  G(p6=v8,p4=v5,p1=v2)>=1
                // 需要将这个用户之前的所有行为全部拿到，来判断是否满足条件，用 state来存储历史数据
                Iterable<LogBean> logBeans = event_state.get();
                int Ucount = 0;
                int Gcount = 0;

                for (LogBean bean : logBeans) {
                    //计算 U事件 次数
                    if ("U".equals(bean.getEventId())) {
                        Map<String, String> properties = bean.getProperties();
                        String p1 = properties.get("p1");
                        String p2 = properties.get("p2");

                        if ("v3".equals(p1) && "v2".equals(p2)) {
                            Ucount++;
                        }
                    }

                    // 计算G事件次数
                    if ("G".equals(bean.getEventId())) {
                        Map<String, String> properties = bean.getProperties();
                        String p1 = properties.get("p1");
                        String p4 = properties.get("p4");
                        String p6 = properties.get("p6");

                        if ("v2".equals(p1) && "v5".equals(p4) && "v8".equals(p6)) {
                            Gcount++;
                        }
                    }
                }

                if (Ucount >= 3 && Gcount >= 1) {
                    // TODO 判断 行为次序条件：  依次做过：  W(p1=v4) ->   R(p2=v3) -> F
                    //将iterator转换为list
                    ArrayList<LogBean> list = new ArrayList<>();
                    CollectionUtils.addAll(list, logBeans.iterator());

                    int index = -1;
                    for (int i = 0; i < list.size(); i++) {
                        LogBean beani = list.get(i);
                        if ("W".equals(beani.getEventId())) {
                            Map<String, String> properties = beani.getProperties();
                            String p1 = properties.get("p1");
                            if ("v4".equals(p1)) {
                                index = i;
                                break;
                            }
                        }
                    }

                    int index2 = -1;
                    if (index >= 0 && index + 1 < list.size()) {
                        for (int i = 0; i < list.size(); i++) {
                            LogBean beani = list.get(i);
                            if ("R".equals(beani.getEventId())) {
                                Map<String, String> properties = beani.getProperties();
                                String p2 = properties.get("p2");
                                if ("v3".equals(p2)) {
                                    index2 = i;
                                    break;
                                }
                            }
                        }
                    }

                    int index3 = -1;
                    if (index2 >= 0 && index2 + 1 < list.size()) {
                        for (int i = 0; i < list.size(); i++) {
                            LogBean beani = list.get(i);
                            if ("F".equals(beani.getEventId())) {
                                index3 = i;
                                break;
                            }
                        }
                    }

                    if (index3 > -1){
                        ResultBean resultBean = new ResultBean();
                        resultBean.setDeviceId(value.getDeviceId());
                        resultBean.setRuleId("test_rule_1");
                        resultBean.setTimeStamp(value.getTimeStamp());
                        out.collect(resultBean);
                    }
                }
            }
        }
    }
}
